package com.yanggu.flink.datastream_api.sink

import cn.hutool.json.JSONUtil
import com.yanggu.flink.datastream_api.pojo.Event
import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.elasticsearch.{ElasticsearchSinkFunction, RequestIndexer}
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink
import org.apache.http.HttpHost
import org.elasticsearch.client.Requests
import org.elasticsearch.common.xcontent.XContentType

import java.util.Collections

/**
 * Sink输出到Elasticsearch中
 * 使用GET http://ip:9200/clicks/_search可以查询到插入的数据
 * https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/connectors/datastream/elasticsearch/#elasticsearch-sink
 */
object ElasticsearchSinkDemo {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val dataStream = env.fromElements(
      Event("Mary", "./home", 1000L),
      Event("Bob", "./cart", 2000L),
      Event("Alice", "./prod?id=100", 3000L),
      Event("Alice", "./prod?id=200", 3500L),
      Event("Bob", "./prod?id=2", 2500L),
      Event("Alice", "./prod?id=300", 3600L),
      Event("Bob", "./home", 3000L),
      Event("Bob", "./prod?id=1", 2300L),
      Event("Bob", "./prod?id=3", 3300L))

    //核心配置es的ip和端口
    val httpHosts = Collections.singletonList(new HttpHost("192.168.1.144", 9200, "http"))

    //如何处理处理, 一般直接插入即可
    val elasticsearchSinkFunction = new ElasticsearchSinkFunction[Event] {
      override def process(data: Event, runtimeContext: RuntimeContext, requestIndexer: RequestIndexer) = {
        val request = Requests
          .indexRequest()
          .index("clicks")
          .source(JSONUtil.toJsonStr(data), XContentType.JSON)

        requestIndexer.add(request)
      }
    }

    val elasticsearchSink = new ElasticsearchSink.Builder[Event](httpHosts, elasticsearchSinkFunction)
      .build()

    dataStream.addSink(elasticsearchSink)

    env.execute()

  }

}
